package org.sample.springbootcommandlinerunner.web;

import org.sample.springbootcommandlinerunner.web.invocation.InvocationException;
import org.sample.springbootcommandlinerunner.web.invocation.InvocationRequest;
import org.sample.springbootcommandlinerunner.web.invocation.InvocationResponse;
import org.sample.springbootcommandlinerunner.web.model.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClientRequestException;
import org.springframework.web.reactive.function.client.WebClientResponseException;
import reactor.core.publisher.Mono;

import java.io.IOException;
import java.nio.file.Path;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.function.Consumer;

@Service
public class WebConnRunner {

	private static final Logger LOG = LoggerFactory.getLogger(WebConnRunner.class);

	private static final int BATCH_SIZE = 300;

	private static final String EXPORT_DIR_PATH = "./conn-client";

	private static final String EXPORT_FILE_NAME_QUEUE_VERSION = "queue_2.3_";

	private static final String EXPORT_FILE_NAME_TAG = "_(no_livechat)";

	@Autowired
	private WebConnService webConnService;

	public void run() {

		LOG.info("Start REST connector test.");

//		runBatchV30(BATCH_SIZE);
		runBatchV31(BATCH_SIZE);
//		runBatchV40(BATCH_SIZE);

		LOG.info("Finished REST connector test");
	}

	private CreateChatRequest buildCreateChatRequest() {
		var request = new CreateChatRequest();
		request.profile(new ChatProfile()
				.pageId("default")
				.pageLocale("en_HK")
				.device("Desktop")
				.app("default")
				.source(null)
				.platform("WEB"))
				.ttl(1800)
				.tags(Collections.emptyList())
				.domain(new ChatDomain()
						.name("VERA")
				)
				.welcome(true)
				.isSessionValid(false);
		return request;
	}

	private RouterCallbackRequest buildRouterCallbackRequest(String session) {
		var request = new RouterCallbackRequest();
		request.session(session)
				.domain(new ChatDomain()
						.name("VERA")
				)
				.messages(Arrays.asList(
						new RouterCallbackMessage()
								.templateKeyType("talkToAgent")
								.templateKeyLanguage("en_hk")
				));
		return request;
	}

	private StartLiveChatRequest buildStartLiveChatRequest(String session, int serialNumber) {
		var request = new StartLiveChatRequest();
		request.session(session)
				.firstName(String.format("Lex%03d", serialNumber))
				.lastName("Chan")
				.title("Test")
				.tier("")
				.lang("en")
				.locale("HK")
				.memberId("")
				.pageId("")
				.intent("")
				.glassboxId("")
				.valueSegment("")
				.keyMarket("")
				.previousConversationId("");
		return request;
	}

	private void runBatchV30(int batchSize) {

		String[] stepDescriptions = new String[]{
				"start",
				"created chat",
				"talk to agent",
				"started livechat",
		};

		var stopWatch = new StopWatch(batchSize, stepDescriptions);

		var latch = new CountDownLatch(batchSize);

		for (int i = 0; i < batchSize; i++) {
			runV30(Mono.just(Integer.valueOf(i)), stopWatch, latch::countDown);
		}

		try {
			latch.await();
		} catch (InterruptedException ex) {
			LOG.error("Latch is interrupted", ex);
			Thread.currentThread().interrupt();
		}

		try {
			LOG.info("Save to CSV");
			stopWatch.saveAsCsv(Path.of(EXPORT_DIR_PATH, String.format("rest_3.0_batch_%d.csv", batchSize)).toString());
		} catch (IOException ex) {
			LOG.error("Save CSV failed", ex);
		}

		clearLiveChatSessions(stopWatch);
	}

	private void runV30(Mono<Integer> serialNumMono, final StopWatch stopWatch, final Runnable done) {

		Mono<InvocationRequest<CreateChatRequest>> createChatReqMono = serialNumMono
				.map(serialNum -> {
					stopWatch.recordStepTime(serialNum);
					return new InvocationRequest<CreateChatRequest>().serialNumber(serialNum).body(buildCreateChatRequest());
				});

		Mono<InvocationResponse<CreateChatResponse>> createChatResMono = createChatReqMono
				.flatMap(request -> webConnService.createChatV30(request));

		Mono<InvocationRequest<RouterCallbackRequest>> routerCallbackReqMono = createChatResMono
				.map(response -> {
					var serialNumber = response.serialNumber();
					var session = response.body().data().session();
					stopWatch.setTaskSessionId(serialNumber, session);
					stopWatch.recordStepTime(serialNumber, "session: " + session);
					return new InvocationRequest<RouterCallbackRequest>()
							.serialNumber(serialNumber).body(buildRouterCallbackRequest(session));
				});

		Mono<InvocationResponse<RouterCallbackResponse>> routerCallbackResMono = routerCallbackReqMono
				.flatMap(request -> webConnService.routerCallbackV30(request));

		Mono<InvocationRequest<StartLiveChatRequest>> startLiveChatReqFlux = routerCallbackResMono
				.map(response -> {
					var serialNumber = response.serialNumber();
					var session = stopWatch.getTaskSessionId(serialNumber);
					stopWatch.recordStepTime(serialNumber);
					return new InvocationRequest<StartLiveChatRequest>()
							.serialNumber(serialNumber).body(buildStartLiveChatRequest(session, serialNumber));
				});

		Mono<InvocationResponse<StartLiveChatResponse>> startLiveChatResMono = startLiveChatReqFlux
				.flatMap(request -> webConnService.startLiveChatV30(request));

		Consumer<InvocationResponse<StartLiveChatResponse>> startLiveChatResConsumer = response -> {
			var serialNumber = response.serialNumber();
			var conversation = response.body().livechatSession();
			stopWatch.setTaskConversationId(serialNumber, conversation);
			stopWatch.recordStepTime(serialNumber, "conversation: " + conversation);
		};

		Consumer<Throwable> errorConsumer = ex -> {
			if (ex instanceof InvocationException) {
				var invocationEx = (InvocationException) ex;
				var error = invocationEx.body().errors().stream()
						.map(ErrorInfo::detail).findFirst().orElse("");
				stopWatch.recordStepError(invocationEx.serialNumber(), error);
			} else {
				LOG.error("Unknown error", ex);
			}
			done.run();
		};

		Runnable completeConsumer = done;

		startLiveChatResMono.subscribe(startLiveChatResConsumer, errorConsumer, completeConsumer);
	}

	private void clearLiveChatSessions(StopWatch stopWatch) {

		var batchSize = stopWatch.taskCount();
		var latch = new CountDownLatch(stopWatch.taskCount());

		for (int i = 0; i < batchSize; i++) {
			Mono.just(i)
					.filter(serialNumber ->
						stopWatch.getTaskSessionId(serialNumber) != null &&
								stopWatch.getTaskConversationId(serialNumber) != null
					)
					.map(serialNumber -> {
						var session = stopWatch.getTaskSessionId(serialNumber);
						Map<String, String> headers = new HashMap<>();
						headers.put("session", session);
						return new InvocationRequest<Void>().serialNumber(serialNumber).headers(headers);
					})
					.flatMap(request -> webConnService.endLiveChatV30(request))
					.subscribe(
							response -> {
								var serialNumber = response.serialNumber();
								LOG.debug("[{}] Ended LiveChat session", serialNumber);
							},
							ex -> {
								if (ex instanceof InvocationException) {
									var invocationEx = (InvocationException) ex;
									var error = invocationEx.body().errors().stream()
											.map(ErrorInfo::detail).findFirst().orElse("");
									LOG.warn("[{}] End livechat session failed. {}", invocationEx.serialNumber(), error);
								} else {
									LOG.error("Unknown error", ex);
								}
								latch.countDown();
							},
							latch::countDown
					);
		}

		try {
			latch.await();
		} catch (InterruptedException ex) {
			LOG.error("Latch is interrupted", ex);
			Thread.currentThread().interrupt();
		}
	}

	private void runBatchV31(int batchSize) {

		String[] stepDescriptions = new String[]{
				"start",
				"created chat",
				"talk to agent",
				"started livechat",
		};

		var stopWatch = new StopWatch(batchSize, stepDescriptions);

		var latch = new CountDownLatch(batchSize);

		for (int i = 0; i < batchSize; i++) {
			runV31(Mono.just(Integer.valueOf(i)), stopWatch, latch::countDown);
		}

		try {
			latch.await();
		} catch (InterruptedException ex) {
			LOG.error("Latch is interrupted", ex);
			Thread.currentThread().interrupt();
		}

		try {
			LOG.info("Save to CSV");
			stopWatch.saveAsCsv(Path.of(EXPORT_DIR_PATH,
					String.format("rest_3.1_%sbatch_%d%s.csv", EXPORT_FILE_NAME_QUEUE_VERSION, batchSize, EXPORT_FILE_NAME_TAG)).toString());
		} catch (IOException ex) {
			LOG.error("Save CSV failed", ex);
		}

		clearLiveChatSessions(stopWatch);
	}

	private void runV31(Mono<Integer> serialNumMono, final StopWatch stopWatch, final Runnable done) {

		/*
		Create chat
		 */
		Mono<InvocationRequest<CreateChatRequest>> createChatReqMono = serialNumMono
				.map(serialNum -> {
					stopWatch.recordStepTime(serialNum);
					return new InvocationRequest<CreateChatRequest>().serialNumber(serialNum).body(buildCreateChatRequest());
				});

		Mono<InvocationResponse<CreateChatResponse>> createChatResMono = createChatReqMono
				.flatMap(request -> webConnService.createChatV31(request));

//		/*
//		Router callback
//		 */
//		Mono<InvocationRequest<RouterCallbackRequest>> routerCallbackReqMono = createChatResMono
//				.map(response -> {
//					var serialNumber = response.serialNumber();
//					var session = response.body().data().session();
//					stopWatch.setTaskSessionId(serialNumber, session);
//					stopWatch.recordStepTime(serialNumber, "session: " + session);
//					return new InvocationRequest<RouterCallbackRequest>()
//							.serialNumber(serialNumber).body(buildRouterCallbackRequest(session));
//				});
//
//		Mono<InvocationResponse<RouterCallbackResponse>> routerCallbackResMono = routerCallbackReqMono
//				.flatMap(request -> restConnService.routerCallbackV30(request));
//
//		/*
//		Start LiveChat
//		 */
//		Mono<InvocationRequest<StartLiveChatRequest>> startLiveChatReqFlux = routerCallbackResMono
//				.map(response -> {
//					var serialNumber = response.serialNumber();
//					var session = stopWatch.getTaskSessionId(serialNumber);
//					stopWatch.recordStepTime(serialNumber);
//					return new InvocationRequest<StartLiveChatRequest>()
//							.serialNumber(serialNumber).body(buildStartLiveChatRequest(session, serialNumber));
//				});
//
//		Mono<InvocationResponse<StartLiveChatResponse>> startLiveChatResMono = startLiveChatReqFlux
//				.flatMap(request -> restConnService.startLiveChatV31(request));


		Consumer<InvocationResponse<CreateChatResponse>> createChatResConsumer = response -> {
			var serialNumber = response.serialNumber();
			var session = response.body().data().session();
			stopWatch.setTaskSessionId(serialNumber, session);
			stopWatch.recordStepTime(serialNumber, "session: " + session);
		};

//		Consumer<InvocationResponse<StartLiveChatResponse>> startLiveChatResConsumer = response -> {
//			var serialNumber = response.serialNumber();
//			var conversation = response.body().livechatSession();
//			stopWatch.setTaskConversationId(serialNumber, conversation);
//			stopWatch.recordStepTime(serialNumber, "conversation: " + conversation);
//		};

		Consumer<Throwable> errorConsumer = ex -> {
			if (ex instanceof InvocationException) {
				var invocationEx = (InvocationException) ex;

				if (invocationEx.body() != null) {
					var error = invocationEx.body().errors().stream()
							.map(ErrorInfo::detail).findFirst().orElse("");
					stopWatch.recordStepError(invocationEx.serialNumber(), error);
				} else if (invocationEx.exception() instanceof WebClientResponseException) {
					var responseException = (WebClientResponseException) invocationEx.exception();
					stopWatch.recordStepError(invocationEx.serialNumber(),
							"Response exception " + responseException.getStatusCode() + " " + responseException.getCause());
				} else if (invocationEx.exception() instanceof WebClientRequestException) {
					var requestException = (WebClientRequestException) invocationEx.exception();
					stopWatch.recordStepError(invocationEx.serialNumber(), "Request exception " + requestException.getCause());
				}
			} else {
				LOG.error("Unknown error", ex);
			}
			done.run();
		};

		Runnable completeConsumer = done;

		createChatResMono.subscribe(createChatResConsumer, errorConsumer, completeConsumer);
//		startLiveChatResMono.subscribe(startLiveChatResConsumer, errorConsumer, completeConsumer);
	}

	private void runBatchV40(int batchSize) {

		String[] stepDescriptions = new String[]{
				"start",
				"creating chat",
				"created chat",
				"talk to agent",
				"starting livechat",
				"started livechat",
		};

		var stopWatch = new StopWatch(batchSize, stepDescriptions);

		var latch = new CountDownLatch(batchSize);

		for (int i = 0; i < batchSize; i++) {
			runV40(Mono.just(Integer.valueOf(i)), stopWatch, latch::countDown);
		}

		try {
			latch.await();
		} catch (InterruptedException ex) {
			LOG.error("Latch is interrupted", ex);
			Thread.currentThread().interrupt();
		}

		try {
			LOG.info("Save to CSV");
			stopWatch.saveAsCsv(Path.of(EXPORT_DIR_PATH,
					String.format("rest_4.0_%sbatch_%d%s.csv", EXPORT_FILE_NAME_QUEUE_VERSION, batchSize, EXPORT_FILE_NAME_TAG)).toString());
		} catch (IOException ex) {
			LOG.error("Save CSV failed", ex);
		}

		clearLiveChatSessions(stopWatch);
	}

	private void runV40(Mono<Integer> serialNumMono, final StopWatch stopWatch, final Runnable done) {

		/*
		Create chat
		 */
		Mono<InvocationRequest<CreateChatRequest>> creatingChatReqMono = serialNumMono
				.map(serialNum -> {
					stopWatch.recordStepTime(serialNum);
					return new InvocationRequest<CreateChatRequest>().serialNumber(serialNum).body(buildCreateChatRequest());
				});

		Mono<InvocationResponse<CreateChatResponse>> creatingChatResMono = creatingChatReqMono
				.flatMap(request -> webConnService.createChatV40(request));

		Mono<InvocationRequest<CreateChatRequest>> createdChatReqMono = creatingChatResMono
				.map(response -> {
					stopWatch.recordStepTime(response.serialNumber());
					var sequence = response.body().sequence();
					Map<String, String> variables = new HashMap<>();
					variables.put("sequence", sequence);
					return new InvocationRequest<CreateChatRequest>().serialNumber(response.serialNumber()).pathVariables(variables);
				});

		Mono<InvocationResponse<CreateChatResponse>> createdChatResMono = createdChatReqMono
				.flatMap(request -> webConnService.createChatV40Result(request));

//		/*
//		Router callback
//		 */
//		Mono<InvocationRequest<RouterCallbackRequest>> routerCallbackReqMono = createdChatResMono
//				.map(response -> {
//					var serialNumber = response.serialNumber();
//					var session = response.body().data().session();
//					stopWatch.setTaskSessionId(serialNumber, session);
//					stopWatch.recordStepTime(serialNumber, "session: " + session);
//					return new InvocationRequest<RouterCallbackRequest>()
//							.serialNumber(serialNumber).body(buildRouterCallbackRequest(session));
//				});
//
//		Mono<InvocationResponse<RouterCallbackResponse>> routerCallbackResMono = routerCallbackReqMono
//				.flatMap(request -> restConnService.routerCallbackV30(request));
//
//		/*
//		Start LiveChat
//		 */
//		Mono<InvocationRequest<StartLiveChatRequest>> startingLiveChatReqMono = routerCallbackResMono
//				.map(response -> {
//					var serialNumber = response.serialNumber();
//					var session = stopWatch.getTaskSessionId(serialNumber);
//					stopWatch.recordStepTime(serialNumber);
//					return new InvocationRequest<StartLiveChatRequest>()
//							.serialNumber(serialNumber).body(buildStartLiveChatRequest(session, serialNumber));
//				});
//
//		Mono<InvocationResponse<StartLiveChatResponse>> startingLiveChatResMono = startingLiveChatReqMono
//				.flatMap(request -> restConnService.startLiveChatV40(request));
//
//
//		Mono<InvocationRequest<StartLiveChatRequest>> startedLiveChatReqMono = startingLiveChatResMono
//				.map(response -> {
//					stopWatch.recordStepTime(response.serialNumber());
//					var sequence = response.body().sequence();
//					Map<String, String> variables = new HashMap<>();
//					variables.put("sequence", sequence);
//					return new InvocationRequest<StartLiveChatRequest>().serialNumber(response.serialNumber()).pathVariables(variables);
//				});
//
//		Mono<InvocationResponse<StartLiveChatResponse>> startedLiveChatResMono = startedLiveChatReqMono
//				.flatMap(request -> restConnService.startLiveChatV40Result(request));


		Consumer<InvocationResponse<CreateChatResponse>> createChatResConsumer = response -> {
			var serialNumber = response.serialNumber();
			var session = response.body().data().session();
			stopWatch.setTaskSessionId(serialNumber, session);
			stopWatch.recordStepTime(serialNumber, "session: " + session);
		};

//		Consumer<InvocationResponse<StartLiveChatResponse>> startLiveChatResConsumer = response -> {
//			var serialNumber = response.serialNumber();
//			var conversation = response.body().livechatSession();
//			stopWatch.setTaskConversationId(serialNumber, conversation);
//			stopWatch.recordStepTime(serialNumber, "conversation: " + conversation);
//		};

		Consumer<Throwable> errorConsumer = ex -> {
			if (ex instanceof InvocationException) {
				var invocationEx = (InvocationException) ex;
				var error = invocationEx.body().errors().stream()
						.map(ErrorInfo::detail).findFirst().orElse(ex.getMessage());
				stopWatch.recordStepError(invocationEx.serialNumber(), error);
			} else {
				LOG.error("Unknown error", ex);
			}
			done.run();
		};

		Runnable completeConsumer = done;

		createdChatResMono.subscribe(createChatResConsumer, errorConsumer, completeConsumer);
//		startedLiveChatResMono.subscribe(startLiveChatResConsumer, errorConsumer, completeConsumer);
	}

}
